package cn.wjssl.trigger.job.springJob;

import cn.bugstack.middleware.db.router.strategy.IDBRouterStrategy;
import cn.wjssl.domain.task.model.entity.TaskEntity;
import cn.wjssl.domain.task.service.ITaskService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;

import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @Author：lxd
 * @Description 重发存入中奖记录时, 未发送成功的mq消息
 * @Date：2025/2/26
 */

@Slf4j
//@Component
public class SendMessageTaskJob {

    @Resource
    private ITaskService taskService;
    @Resource
    private ThreadPoolExecutor executor;
    @Resource
    private IDBRouterStrategy dbRouter;

    @Scheduled(cron = "0/5 * * * * ?")
    public void exec() {
        log.info("定时任务，【重发 MQ 消息】");
        try {
            // 获取分库数量
            int dbCount = dbRouter.dbCount();
            // 扫描所有的分库
            for (int dbIdx = 1; dbIdx <= dbCount; dbIdx++) {
                int finalDbIdx = dbIdx;
                /**
                 * 多线程执行
                 * 1. 外层线程：
                 *    每个数据库分片（dbIdx）会开启一个线程去处理对应数据库的任务。
                 *    目的是并发扫描多个数据库，加快任务查询速度。
                 * 2. 内层线程：
                 *    对每个查询出来的 发送失败的task 再开启一个线程去异步发送 MQ 消息。
                 *    这样可以并行处理单个数据库内的多个任务，加快任务发送速度。
                 *
                 * 使用线程池策略 CallerRunsPolicy 可以在线程池资源紧张时让主线程执行任务，避免因线程池满了而导致任务无法及时发送。
                 *
                 * 如果 executor.execute() 放在for循坏外, 那就相当于只有一个线程在执行for循环, 还是串行
                 * 放在for循环里面, 每次循坏都会创建一个新的任务交给线程池, 线程池会并发执行这些任务,
                 * 因为 executor.execute() 提交的是一个异步任务, 所以任务提交后, 会立即返回, 不需要等里面的任务执行完, 不会阻塞当前线程继续执行下一轮循环。
                 */
                executor.execute(() -> {
                    // 设置查询哪个分库
                    dbRouter.setDBKey(finalDbIdx);
                    // 设置查询哪个分表, 因为task只有一张表, 所以索引是0
                    dbRouter.setTBKey(0);
                    // 查询发送MQ失败和超时1分钟未发送的MQ limit 10
                    List<TaskEntity> taskEntities = taskService.queryNoSendMessageTaskList();
                    // 如何没有发送失败的, 那直接返回, 等待下一次定时任务
                    if (taskEntities.isEmpty()) return;
                    for (TaskEntity taskEntity : taskEntities) {
                        // 再开启一个线程去异步重发 MQ 消息。
                        executor.execute(() -> {
                            try {
                                // 重发任务
                                taskService.sendMessage(taskEntity);
                                // 发送成功, 修改为completed
                                taskService.updateTaskSendMessageCompleted(taskEntity.getUserId(), taskEntity.getMessageId());
                            }catch (Exception e) {
                                log.error("定时任务，发送MQ消息失败 userId: {} topic: {}", taskEntity.getUserId(), taskEntity.getTopic());
                                // 那还是设置该任务为fail, 等待下次发送
                                taskService.updateTaskSendMessageFail(taskEntity.getUserId(), taskEntity.getMessageId());
                            }
                        });
                    }
                });
            }

        }catch (Exception e) {
            log.error("定时任务，扫描MQ任务表发送消息失败。", e);
        } finally {
            dbRouter.clear();
        }
    }
}
